package cn.cup.dmp.report

import java.util.Properties

import cn.cup.dmp.beans.RptProvinceCityResult
import cn.cup.dmp.config.ConfigHelper
import com.google.gson.Gson
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}
import scalikejdbc.{DB, SQL}

object Province_City_Count_CoreStyle {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setAppName("DMP平台").setMaster("local[*]")
    conf.set("spark.serializer", ConfigHelper.ser)
    //设置序列化方式
    val sc = new SparkContext(conf)
    val sql = new SQLContext(sc)

    //读取parquet文件
    val dataFrame: DataFrame = sql.read.parquet(ConfigHelper.destPath)
    //逐行读取数据，提取省市信息
    val value: RDD[((String, String), Int)] = dataFrame.map(row => {
      val province = row.getAs[String]("provincename")
      val city = row.getAs[String]("cityname")
      ((province, city), 1)
    }).reduceByKey(_ + _)

    //结果保存为Json文件
    //方式1
    val rptRest: RDD[String] = value.map(m => {
      val gson = new Gson()//GSON没有实现序列化，不能跨机器使用
      gson.toJson(RptProvinceCityResult(m._1._1,m._1._2,m._2))//String 实现了序列化
    })
    rptRest.saveAsTextFile("D:\\小牛视频\\资料_广告项目\\省市统计表")

   // rptRest.map(t => gson.toJson(t)).foreach(println)

    //方式2 转化为DataFrame，保存为json文件
    import sql.implicits._
    val res: DataFrame = value.map(m =>(m._1._1,m._1._2,m._2)).toDF("province","city","ct")
    res.write.json("D:\\小牛视频\\资料_广告项目\\省市统计表")

    //数据写入Mysql   方式1 sparkSql
    val connetProperties = new Properties()
    connetProperties.setProperty("user","root")
    connetProperties.setProperty("password","123456")
    res.write.jdbc(ConfigHelper.url,ConfigHelper.tableName,connetProperties)

    //方式2 ScalikeJDBC
    ConfigHelper.dbSetup
    value.foreachPartition(par => {
      DB.localTx(implicit session =>{
        par.foreach(m =>{
          SQL("insert into rpt_province_city_count2(province,city,ct) values(?,?,?)").bind(m._1._1,m._1._2,m._2).update().apply()
        })
      })
    })

    sc.stop()

  }

}
